

package com.coding4play.mqtt.broker.protocol;

import com.coding4play.mqtt.broker.config.BrokerProperties;
import com.coding4play.mqtt.common.entity.session.ChannelSession;
import com.coding4play.mqtt.common.repository.ISessionRepository;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * PINGREQ连接处理
 */
@AllArgsConstructor
@Slf4j
public class PingReq {

    private ISessionRepository sessionRepository;
    private BrokerProperties brokerProperties;

    public void processPingReq(Channel channel, MqttMessage msg) {
        String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
        if (sessionRepository.hasSession(clientId)) {
            ChannelSession channelSession = sessionRepository.get(clientId);
//            ChannelId channelId = channelIdMap.get(sessionStore.getBrokerId() + "_" + sessionStore.getChannelId());
            Channel c0 = this.sessionRepository.getChannel(channelSession);
            if (brokerProperties.getId().equals(channelSession.getBrokerId()) && c0 != null) {
                sessionRepository.expire(clientId, channelSession.getExpire());
                MqttMessage pingRespMessage = MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0), null, null);
                log.debug("PINGREQ - clientId: {}", clientId);
                channel.writeAndFlush(pingRespMessage);
            }
        }
    }

}
